package com.hub.realtime.flink.submit.impl;


import java.lang.reflect.Constructor;
import java.net.URI;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Executors;

import com.hub.realtime.common.utils.ListUtils;
import io.fabric8.kubernetes.client.NamespacedKubernetesClient;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Options;
import org.apache.flink.client.cli.*;
import org.apache.flink.client.deployment.*;
import org.apache.flink.client.deployment.application.ApplicationConfiguration;
import org.apache.flink.client.deployment.application.cli.ApplicationClusterDeployer;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.ClusterClientProvider;
import org.apache.flink.client.program.PackagedProgramUtils;
import org.apache.flink.configuration.*;
import org.apache.flink.kubernetes.KubernetesClusterDescriptor;
import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
import org.apache.flink.kubernetes.configuration.KubernetesDeploymentTarget;
import org.apache.flink.kubernetes.kubeclient.Fabric8FlinkKubeClient;
import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;

import io.fabric8.kubernetes.client.DefaultKubernetesClient;
import org.apache.flink.runtime.security.SecurityConfiguration;
import org.apache.flink.runtime.security.SecurityUtils;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
import org.apache.hadoop.yarn.api.records.ApplicationId;

import static org.apache.flink.util.Preconditions.checkNotNull;


public class FlinkOnK8sApplicationSubmit {
    public static void main(String[] args) throws Exception {

        String confDir="/Users/admin/softapp/flink-1.13.2/conf";
        final Configuration defaultConfig=GlobalConfiguration.loadConfiguration(confDir);
        List<CustomCommandLine> customCommandLines = CliFrontend.loadCustomCommandLines(defaultConfig, confDir);
        final CliFrontend cliFrontend = new CliFrontend(defaultConfig, customCommandLines);
        SecurityUtils.install(new SecurityConfiguration(defaultConfig));
        //获取flink所有的命令行参数列表，对args做解析
        final Options commandOptions = CliFrontendParser.getRunCommandOptions();
        //构建args
//KubernetesDeploymentTarget
        List<String>runArgs=new ArrayList<>();
        runArgs.add("-t"+KubernetesDeploymentTarget.APPLICATION.getName());
        runArgs.add("-jlocal:///opt/flink/jars/myflinktest-1.0-SNAPSHOT.jar");
        runArgs.add("-ccom.wp.flinkdemo.FlinkTestDemo");
        runArgs.add("-n");
        runArgs.add("-p1");
       // runArgs.add("-s/opt/sss");
        //runArgs.add("-d");

        String[] finalArgs = ListUtils.toArray(runArgs, String.class);

        final CommandLine commandLine = cliFrontend.getCommandLine(commandOptions, finalArgs, true);
        CustomCommandLine activeCommandLine =
                cliFrontend.validateAndGetActiveCommandLine(checkNotNull(commandLine));

        DefaultClusterClientServiceLoader clusterClientServiceLoader= new DefaultClusterClientServiceLoader();

        ApplicationDeployer deployer =
                new ApplicationClusterDeployer(clusterClientServiceLoader);
        Class<ProgramOptions> programOptionsClass = ProgramOptions.class;
        Constructor<ProgramOptions> constructor = programOptionsClass.getDeclaredConstructor(CommandLine.class);
        constructor.setAccessible(true);
        ProgramOptions programOptions= constructor.newInstance(commandLine);
        programOptions.validate();
        final URI uri = PackagedProgramUtils.resolveURI(programOptions.getJarFilePath());
        Configuration effectiveConfiguration = getEffectiveConfiguration(activeCommandLine, commandLine, programOptions, Collections.singletonList(uri.toString()),defaultConfig);

        //设置main的args
        List<String>mainArgs=new ArrayList<>();
        mainArgs.add("--host");
        mainArgs.add("2522222");

        if (!mainArgs.isEmpty()) {
            effectiveConfiguration.set(ApplicationConfiguration.APPLICATION_ARGS, mainArgs);
        }

        //设置其他参数配置参数


        effectiveConfiguration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY,MemorySize.parse("2048M"));
        effectiveConfiguration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY,MemorySize.parse("2048M"));

        effectiveConfiguration.set(ApplicationConfiguration.APPLICATION_MAIN_CLASS, programOptions.getEntryPointClassName());
        effectiveConfiguration.setString("kubernetes.jobmanager.service-account","flink-service-account");



        //k8s 设置

        effectiveConfiguration .set(KubernetesConfigOptions.NAMESPACE, "kube-public")
                .set(KubernetesConfigOptions.CLUSTER_ID, "jobmanager-"+ Instant.now().toEpochMilli())
                .set(KubernetesConfigOptions.CONTAINER_IMAGE, "repoin.shizhuang-inc.net/flink_test/flink-k8s-test:v1")
                .set(KubernetesConfigOptions.REST_SERVICE_EXPOSED_TYPE, KubernetesConfigOptions.ServiceExposedType.ClusterIP)
                .set(KubernetesConfigOptions.CONTAINER_IMAGE_PULL_POLICY, KubernetesConfigOptions.ImagePullPolicy.IfNotPresent);
                //.set(DeploymentOptions.ATTACHED, false)
                //.set(CheckpointingOptions.INCREMENTAL_CHECKPOINTS, true);
                //.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.parse("1024M"))
              //  .set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.parse("1024M"))
              //  .set(CheckpointingOptions.MAX_RETAINED_CHECKPOINTS, 2);




        ApplicationConfiguration applicationConfiguration = ApplicationConfiguration.fromConfiguration(effectiveConfiguration);


        ClusterClientFactory<ApplicationId> clusterClientFactory = clusterClientServiceLoader.getClusterClientFactory(effectiveConfiguration);
        ClusterDescriptor<ApplicationId> clusterDescriptor = clusterClientFactory.createClusterDescriptor(effectiveConfiguration);


        ClusterSpecification clusterSpecification =
                clusterClientFactory.getClusterSpecification(effectiveConfiguration);

        ClusterClientProvider<ApplicationId> applicationIdClusterClientProvider = clusterDescriptor.deployApplicationCluster(
                clusterSpecification, applicationConfiguration);

        ClusterClient<ApplicationId> clusterClient = applicationIdClusterClientProvider.getClusterClient();

        System.out.println(clusterClient.getClusterId());


        //deployer.run(effectiveConfiguration, applicationConfiguration);
        // configuration.set(DeploymentOptionsInternal.CONF_DIR, flinkRunRequest.getFlinkHome().concat("/conf"));

        System.out.println(effectiveConfiguration);

    }





    private static<T> Configuration getEffectiveConfiguration(
            final CustomCommandLine activeCustomCommandLine,
            final CommandLine commandLine,
            final ProgramOptions programOptions,
            final List<T> jobJars,Configuration defaultConfig) throws FlinkException {



        final Configuration effectiveConfiguration =
                getEffectiveConfiguration(activeCustomCommandLine, commandLine,defaultConfig);


        final ExecutionConfigAccessor executionParameters =
                ExecutionConfigAccessor.fromProgramOptions(
                        checkNotNull(programOptions), checkNotNull(jobJars));


        executionParameters.applyToConfiguration(effectiveConfiguration);


        return effectiveConfiguration;

    }



    private static <T>Configuration getEffectiveConfiguration(
            final CustomCommandLine activeCustomCommandLine, final CommandLine commandLine,Configuration defaultConfig)
            throws FlinkException{
        final Configuration effectiveConfiguration = new Configuration(defaultConfig);
        final Configuration commandLineConfiguration =  Preconditions.checkNotNull(activeCustomCommandLine).toConfiguration(commandLine);
        effectiveConfiguration.addAll(commandLineConfiguration);
        return effectiveConfiguration;
    }



}